package com.xxl.job.executor.service.jobhandler;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.common.util.concurrent.*;
import com.ruoyi.bizsys.domain.SmsTask;
import com.ruoyi.bizsys.domain.SmsTemplate;
import com.ruoyi.bizsys.service.ISmsTaskService;
import com.ruoyi.bizsys.service.ISmsTemplateService;
import com.ruoyi.bizsys.utils.BizConstants;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.common.utils.Func;
import com.ruoyi.sms.service.ISmsService;
import com.ruoyi.system.service.ISysConfigService;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.executor.mapper.QuartzSmsMapper;
import com.xxl.job.executor.service.thread.SmsTaskThread;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * 短信发送任务
 *
 * @author ruoyi
 */
@Slf4j
@AllArgsConstructor
@Component
public class SmsSendJob {

    private ISmsTaskService smsTaskService;

    private ISmsService smsService;

    private QuartzSmsMapper quartzSmsMapper;

    private ISmsTemplateService smsTemplateService;

    private ISysConfigService sysConfigService;

    private RedisCache redisCache;

    @XxlJob("smsSendJobHandler")
    public void execute() {

        XxlJobHelper.log("短信发送任务 Start >>>>>>>>>>");

        // 只发送当前时刻的短信模版
        String oClock = DateUtils.getCurrentClock();
        XxlJobHelper.log("当前时刻：{}", oClock);

        List<String> smsTypes = new ArrayList<>();
        List<SmsTemplate> smsTemplates = smsTemplateService.list(new QueryWrapper<SmsTemplate>().select("sms_type", "send_clock").eq("state", BizConstants.BIZ_STATUS_1));
        smsTemplates.forEach(e -> {
            //单个模版可以选择多时刻发送，空则默认全天
            if (Func.isNotEmpty(e.getSendClock())) {
                String[] times = e.getSendClock().split(",");
                if (Arrays.asList(times).contains(oClock)) {
                    smsTypes.add(e.getSmsType());
                }
            } else {
                smsTypes.add(e.getSmsType());
            }
        });

        // 区分普通短信和定期短信
        List<SmsTask> smsTaskList = quartzSmsMapper.getSmsTaskInType(smsTypes);
        int smsTaskListSize = smsTaskList.size();
        if (smsTaskListSize <= 0) {
            XxlJobHelper.log("短信发送任务 End >>>>>>>>>> 无待处理的短信任务");
            return;
        }

        XxlJobHelper.log("待处理短信任务记录数：{}", smsTaskList.size());

        Date date = new Date();
        String redisName = "SmsSendTask-" + DateUtils.dateTime(date);

        //每个线程执行的客户数
        String smsSendTaskConfig = sysConfigService.selectConfigByKey("sms_send_task_config");
        String[] split = smsSendTaskConfig.split(",");
        int threadStep = Integer.parseInt(split[0]);

        //需要multiple个线程来执行
        int threadCount = smsTaskListSize / threadStep;
        if (threadCount * threadStep != smsTaskListSize) {
            threadCount++;
        }

        //线程池大小
        int poolSize = Integer.parseInt(split[1]);
        ExecutorService executorService = Executors.newFixedThreadPool(poolSize);
        ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(executorService);

        //线程计数器
        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        int threadAccountCount = 0;
        while (threadAccountCount != smsTaskListSize) {
            //分页取数startNum
            int startLimit = threadAccountCount;

            threadAccountCount += threadStep;
            if (threadAccountCount > smsTaskListSize) {
                threadAccountCount = smsTaskListSize;
            }

            String splitRedisName = redisName + ":" + startLimit + "-" + threadAccountCount;
            List<SmsTask> splitSmsTaskList = smsTaskList.subList(startLimit, threadAccountCount);
            redisCache.setCacheList(splitRedisName, splitSmsTaskList);
            redisCache.expire(splitRedisName, 23, TimeUnit.HOURS);

            SmsTaskThread thread = new SmsTaskThread("[SmsTaskThread-" + startLimit + "]", splitRedisName, smsService, quartzSmsMapper, smsTaskService, redisCache);
            ListenableFuture listenableFuture = listeningExecutorService.submit(thread);

            Futures.addCallback(listenableFuture, new FutureCallback<String>() {
                @Override
                public void onSuccess(String result) {
                    countDownLatch.countDown();
                    XxlJobHelper.log("[" + this.getClass().getName() + "]线程处理结果：" + result);
                }

                @Override
                public void onFailure(Throwable throwable) {
                    countDownLatch.countDown();
                    XxlJobHelper.log("[" + this.getClass().getName() + "]线程处理出错：" + throwable);
                }
            });
        }

        try {
            countDownLatch.await(3 * 60, TimeUnit.MINUTES);//超过3个小时放弃
        } catch (InterruptedException e) {
            XxlJobHelper.log("短信发送任务执行失败，非正常结束", e);
            XxlJobHelper.handleFail("短信发送任务执行失败");
        }

        XxlJobHelper.log("短信发送任务 End >>>>>>>>>>");
    }

}
